package com;

import io.debezium.data.Json;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.relational.history.FileDatabaseHistory;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) {
        // 配置
        Properties props = new Properties();
        props.setProperty("name", "oracle-engine-0036"); //唯一实例
        props.setProperty("connector.class", OracleConnector.class.getName());
        props.setProperty("offset.storage", FileOffsetBackingStore.class.getName());
        // 指定 offset 存储目录
        props.setProperty("offset.storage.file.filename", "D:/data/oracledebezium/oracle4.txt");
        // 指定 Topic offset 写入磁盘的间隔时间
        props.setProperty("offset.flush.interval.ms", "60000");
        //设置数据库连接信息
        props.setProperty("database.hostname", "10.69.72.112");
        props.setProperty("database.port", "11522");
        props.setProperty("database.user", "source1");
        props.setProperty("database.password", "oracle");
        props.setProperty("database.server.id", "85701");
        //C##DBZUSER.STU可以用正则
        //props.setProperty("table.include.list", "STUDENT1");
        props.setProperty("snapshot.include.collection.list", "C##SOURCE1");
        props.setProperty("database.history",
                FileDatabaseHistory.class.getCanonicalName());
        props.setProperty("database.history.file.filename",
                "D:/data/oracledebezium/oracle5.txt");
        //每次运行需要对此参数进行修改，因为此参数唯一
        props.setProperty("database.server.name", "my-oracle-connector-0026");
        //指定 CDB 模式的实例名
        props.setProperty("database.dbname", "zujian4");
        //是否输出 schema 信息
        props.setProperty("key.converter.schemas.enable", "false");
        props.setProperty("value.converter.schemas.enable", "false");
        props.setProperty("database.serverTimezone", "UTC"); // 时区
        props.setProperty("database.connection.adapter", "logminer"); // 模式


        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);
        // 2. 构建 DebeziumEngine
        // 使用 Json 格式
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine
                .create(Json.class)
                .using(props)
                .notifying(record -> {
                    // record中会有操作的类型（增、删、改）和具体的数据
                    System.out.println(record);
                    System.out.println("record.key() = " + record.key());
                    System.out.println("record.value() = " + record.value());
                })
                .using((success, message, error) -> {
                    //查看错误信息
                    if (!success && error != null) {
                        // 报错回调
                        System.out.println("----------error------");
                        System.out.println(message);
                        System.out.println(error);
                        error.printStackTrace();
                    }
                })
                .build();

        return engine;

    }
}